package io.gbznt.config;

import java.io.BufferedInputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.Date;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;

import io.gbznt.modules.buss.entity.BussRtuInfoEntity;
import io.gbznt.modules.buss.entity.BussRtuRecordEntity;
import io.gbznt.modules.buss.entity.BussRtuSendRecordEntity;
import io.gbznt.modules.buss.service.BussRtuSendRecordService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;

import io.gbznt.common.utils.HexUtil;
import io.gbznt.modules.buss.service.BussRtuInfoService;
import io.gbznt.modules.buss.service.BussRtuRecordService;

/**
 * RTU TCP 2088 端口启动
 * 
 * @author Administrator
 *
 */
@Component
@Order(value = 1)
public class TcpStartService implements ApplicationRunner {

	private static Logger logger = LoggerFactory.getLogger(TcpStartService.class);

	public static final Integer SOCKET_PORT = 2088;
	/**
	 * RTU 下行队列
	 */
	public static ConcurrentLinkedQueue<BussRtuSendRecordEntity> RTUSEND_LINKED_QUEUE = new ConcurrentLinkedQueue<BussRtuSendRecordEntity>();

	/**
	 * 装上行报文与dos的队列
	 */
	public static ConcurrentLinkedQueue<SxbwAndDos> SXBW_DOS_QUEUE = new ConcurrentLinkedQueue<SxbwAndDos>();
	/**
	 * 连接池
	 */
	public static ConcurrentLinkedQueue<SxbwAndDos> SXBW_DOS_QUEUE_ljc = new ConcurrentLinkedQueue<SxbwAndDos>();
	/**
	 * 定时清除时间(下行)
	 */
	private static final Integer TIME_GC_0 = 30 * 60 * 1000;
	/**
	 * 定时清除时间(上行)
	 */
	private static final Integer TIME_GC_1 = 10 * 60 * 1000;

	@Autowired
	private BussRtuRecordService bussRtuRecordService;
	@Autowired
	private BussRtuSendRecordService bussRtuSendRecordService;
	@Autowired
	private BussRtuInfoService bussRtuInfoService;

	@Override
	public void run(ApplicationArguments args) throws Exception {

		// 启动TCP线程
		new Thread(new TcpRuns()).start();

		// 启动RTU获取下行报文线程
		new Thread(new RtuReadSendRuns()).start();

		// 发送下行报文线程
		new Thread(new RtuReadSendRunsFs()).start();

	}

	/**
	 * Tcp线程执行
	 *
	 * @author Administrator
	 *
	 */
	class TcpRuns implements Runnable {
		@Override
		public void run() {
			logger.info("启动TCP监听端口:" + SOCKET_PORT);

			ServerSocket serverSocket = null;
			try {
				serverSocket = new ServerSocket(SOCKET_PORT);

				// 通过死循环开启长连接，开启线程去处理消息
				while (true) {
					Socket socket = serverSocket.accept();
					new Thread(new TcpSocketRuns(socket)).start();
				}
			} catch (Exception e) {
				e.printStackTrace();
			} finally {
				try {
					if (serverSocket != null) {
						serverSocket.close();
					}
				} catch (Exception e2) {
					e2.printStackTrace();
				}
			}
		}
	}

	/**
	 * RTU读取下行报文线程执行
	 *
	 * @author Administrator
	 *
	 */
	class RtuReadSendRuns implements Runnable {
		@Override
		public void run() {
			logger.info("启动获取RTU下行报文");

			List<BussRtuSendRecordEntity> newTaskList;
			try {

				while (true) {
					// 获取待处理的下行报文
					newTaskList = bussRtuSendRecordService
							.list(new QueryWrapper<BussRtuSendRecordEntity>().eq("status", 0));

					if (null != newTaskList && newTaskList.size() > 0) {
						for (BussRtuSendRecordEntity newTask : newTaskList) {
							RTUSEND_LINKED_QUEUE.offer(newTask);

							// 取出后改变状态为待发送
							newTask.setStatus(1);
							bussRtuSendRecordService.updateById(newTask);
						}

						logger.info("获取到" + newTaskList.size() + "条下行报文");
					}
					// 休眠时间
					Thread.sleep(10000);
				}
			} catch (Exception e) {
				e.printStackTrace();
			}

		}
	}

	/**
	 * 发送下行报文线程
	 * 
	 * @author Administrator
	 *
	 */
	public class RtuReadSendRunsFs implements Runnable {
		@Override
		public void run() {
			logger.info("发送下行报文线程");

			// 下行报文回馈
			while (true) {
				if (TcpStartService.RTUSEND_LINKED_QUEUE.size() == 0) {
					continue;
				}

				// 获取下行报文
				BussRtuSendRecordEntity currTask = TcpStartService.RTUSEND_LINKED_QUEUE.poll();

				// 如果超时，则清除
				if ((new Date()).getTime() - currTask.getCreateTime().getTime() > TIME_GC_0) {
					currTask.setStatus(3);// 状态为已发送失败
					bussRtuSendRecordService.updateById(currTask);// 修改
					continue;
				}

				// 连接池
				flag: for (int i = 0; i < TcpStartService.SXBW_DOS_QUEUE_ljc.size(); i++) {
					// 封装的实体类
					SxbwAndDos sad_1 = TcpStartService.SXBW_DOS_QUEUE_ljc.poll();
					// 流
					DataOutputStream dos_1 = sad_1.getDos();
					boolean flag = TcpStartService.dosWriteIfOpen(dos_1);

					if (flag == false) {// 连接失效
						continue;
					}
					// 塞回去
					TcpStartService.SXBW_DOS_QUEUE_ljc.offer(sad_1);

					// 下行与连接池匹配
					if (!currTask.getRtuId().equals(sad_1.getSxbw().getRtuId())) {
						TcpStartService.RTUSEND_LINKED_QUEUE.offer(currTask);
						continue;
					}

					currTask.setStatus(2);// 状态为已发送
					bussRtuSendRecordService.updateById(currTask);// 修改

					for (int c = 0; c < 3; c++) {
						byte[] btyeSend = TcpStartService.hexStrToByteArrs(currTask.getMessageCode());
						try {
							logger.info("发送下行---------------------");
							dos_1.write(btyeSend, 0, btyeSend.length);
						} catch (IOException e1) {
							logger.info("发送下行失联--------------------");
							e1.printStackTrace();
						}

						try {
							Thread.sleep(200);
						} catch (InterruptedException e) {
							e.printStackTrace();
						}

						int sdq = TcpStartService.SXBW_DOS_QUEUE.size();
						// 用上行报文队列来循环
						for (int j = 0; j < sdq; j++) {
							// 封装的实体类
							SxbwAndDos sad = TcpStartService.SXBW_DOS_QUEUE.poll();
							// 上行报文
							BussRtuRecordEntity sxbw = sad.getSxbw();
							// 如果超时，清除
							if ((new Date()).getTime() - sxbw.getUpdateTime().getTime() > TIME_GC_1) {
								continue;
							}

							// 报文匹配
							if (currTask.getRtuId().equals(sxbw.getRtuId())
									&& currTask.getBaowentype().equals(sxbw.getBaowentype())) {// 如果匹配(rtuId相同且报文特征相同)
								currTask.setStatus(4);// 状态为已发送成功
								bussRtuSendRecordService.updateById(currTask);// 修改
								logger.info("发送下行报文成功");
								break flag;

							} else {// 不匹配就把下行队列放回去，上行队列用过的值放入上行备用队列
								TcpStartService.SXBW_DOS_QUEUE.offer(sad);
							}
						}

						if (c == 2) {
							currTask.setStatus(3);// 状态为已发送失败
							bussRtuSendRecordService.updateById(currTask);// 修改
							break flag;
						}
					}
				}
			}
		}
	}

	public static Boolean dosWriteIfOpen(DataOutputStream dos) {
		try {
			dos.writeInt(1);
			return true;
		} catch (IOException e) {
			return false;
		}

	}

	/**
	 * TcpSocket线程执行
	 * 
	 * @author Administrator
	 *
	 */
	class TcpSocketRuns implements Runnable {

		Socket socket;

		BufferedInputStream bis;
		DataInputStream dis;
		DataOutputStream dos;

		public TcpSocketRuns(Socket socket) {
			super();
			this.socket = socket;
		}

		@Override
		public void run() {
			try {
				// 装饰流BufferedReader封装输入流（接收客户端的流）
				bis = new BufferedInputStream(socket.getInputStream());
				dis = new DataInputStream(bis);
				dos = new DataOutputStream(socket.getOutputStream());

				String lineString = "";
				byte[] b = new byte[1];

				Integer count = 0;// 报文是第几次进入的

				while (dis.read(b) != -1) {
					// 将字节数组转换成十六进制的字符串
					lineString += HexUtil.byte2HexStr(b) + " ";

					if (null == lineString || "bye".equals(lineString)) {
						break;
					}

					// 一个请求
					if (dis.available() == 0) {
						count++;
						logger.info("开始接收报文：" + lineString);

						// 报文数组
						String[] str = lineString.split(" ");

						//先查询是否有rtu
						String string = str[24] + str[25] + str[26] + str[27] + str[28];
						QueryWrapper<BussRtuInfoEntity> rtuqw = new QueryWrapper<>();
						rtuqw.eq("serial_num", string);
						BussRtuInfoEntity rtu = bussRtuInfoService.getOne(rtuqw);
						
						if(rtu==null) {
							continue;
						}
						
						int length = str.length;
						String gongnengma = str[10];// 功能码
						String mima = str[8] + str[9];// 密码
						String zhanghao = str[3] + str[4] + str[5] + str[6] + str[7];// 账号

						String str1 = rtu.getPassword();// 密码匹配值,动态获取
						String str2 = rtu.getSerialNum();// 账号匹配值,动态获取

						// 以下报文都是根据上行报文来判断的
						if (length == 107 && "34".equals(gongnengma) && mima.equals(str1) && zhanghao.equals(str2)) {// 遥测站定时报（AFN=34H）
							logger.info("遥测站定时报----");
							boolean flag = bussRtuRecordService.addRtuRecord(str, 1, lineString, dos, count);
							if (flag == true) {
								logger.info("遥测站定时报报文记录----添加成功");
							} else {
								logger.info("遥测站定时报报文记录----添加失败");
							}
						} else if (length == 64 && "4E".equals(gongnengma) && mima.equals(str1)
								&& zhanghao.equals(str2)) {// 开关闸指令（AFN=4EH）
							logger.info("开关闸指令----");
							boolean flag = bussRtuRecordService.addRtuRecord(str, 2, lineString, dos, count);
							if (flag == true) {
								logger.info("开关闸指令报文记录----添加成功");
							} else {
								logger.info("开关闸指令报文记录----添加失败");
							}
						} else if (length == 70 && "37".equals(gongnengma) && mima.equals(str1)
								&& zhanghao.equals(str2)) {// 中心站查询遥测站实时数据（AFN=37H）
							logger.info("中心站查询遥测站实时数据----");
							boolean flag = bussRtuRecordService.addRtuRecord(str, 3, lineString, dos, count);
							if (flag == true) {
								logger.info("中心站查询遥测站实时数据报文记录----添加成功");
							} else {
								logger.info("中心站查询遥测站实时数据报文记录----添加失败");
							}
						} else if (length == 50 && "43".equals(gongnengma) && mima.equals(str1)
								&& zhanghao.equals(str2)) {// 中心站读取遥测站基本配置表/遥测站自报基本配置表（AFN=43H）
							logger.info("中心站读取遥测站基本配置表/遥测站自报基本配置表----");
							boolean flag = bussRtuRecordService.addRtuRecord(str, 4, lineString, dos, count);
							if (flag == true) {
								logger.info("中心站读取遥测站基本配置表/遥测站自报基本配置表报文记录----添加成功");
							} else {
								System.out.println("中心站读取遥测站基本配置表/遥测站自报基本配置表报文记录----添加失败");
								logger.info("中心站读取遥测站基本配置表/遥测站自报基本配置表报文记录----添加失败");
							}
						} else if (length == 50 && "42".equals(gongnengma) && mima.equals(str1)
								&& zhanghao.equals(str2)) {// 中心站修改遥测站运行参数配置表（AFN=42H）
							logger.info("中心站修改遥测站运行参数配置表----");
							boolean flag = bussRtuRecordService.addRtuRecord(str, 5, lineString, dos, count);
							if (flag == true) {
								logger.info("中心站修改遥测站运行参数配置表报文记录----添加成功");
							} else {
								logger.info("中心站修改遥测站运行参数配置表报文记录----添加失败");
							}
						} else {
							logger.info("报文不匹配!!!");
						}

						lineString = "";
					}

				}

			} catch (Exception e) {
				e.printStackTrace();
			} finally {
				try {
					if (dis != null) {
						dis.close();
					}
					if (bis != null) {
						bis.close();
					}
					if (dos != null) {
						dos.close();
					}

					if (socket != null) {
						socket.close();
					}
				} catch (Exception e2) {
					e2.printStackTrace();
				}
			}
		}

	}

	/**
	 * hex字符串 to hex
	 * 
	 * @param hexString
	 * @return
	 */
	public static byte[] hexStrToByteArrs(String hexString) {
		// if (StringUtils.isEmpty(hexString)) {
		// return null;
		// }

		hexString = hexString.replaceAll(" ", "");
		int len = hexString.length();
		int index = 0;

		byte[] bytes = new byte[len / 2];

		while (index < len) {
			String sub = hexString.substring(index, index + 2);
			bytes[index / 2] = (byte) Integer.parseInt(sub, 16);
			index += 2;
		}

		return bytes;
	}

}
